package com.peng.sparktest.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object SparkApiTest04_Partitions {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("pv_uv_job").setMaster("local")
    val context = new SparkContext(conf)
    context.setLogLevel("ERROR")
    val data: RDD[String] = context.parallelize(List("1", "2", "3", "4", "5", "6", "7", "8"), 2)
    // 实现对一个数据集里的数据进行添加一些额外属性，额外属性可能来自于db查询
    data.mapPartitionsWithIndex((pIndex, pIter) => {
      println(s"connect $pIndex db~~~~")
      val buffer: ListBuffer[String] = new ListBuffer()
      while (pIter.hasNext) {
        val i: String = pIter.next()
        println(s"$i select....")
        buffer.+=(i + "selected")
      }
      println("close db~~~~")
      buffer.iterator
    }).foreach(println)

    //以上的写法有没有什么问题？  当数据量过大时，就会造成OOM的情况出现，
    // 因为正常来说迭代器是一部分一部分拿出数据，但这里为了加工数据，将数据从迭代器取出，加工，然后放入了ListBuffer，
    //这意味着，一个分区的全数据量在此时全都会涌入了内存

    //灵魂的升华--------贯彻全局的迭代器模式  其实就是就是为此种情况准备的，接下来使用迭代器的方式进行实现：
    data.mapPartitionsWithIndex((pIndex, pIter) => {
      new Iterator[String] {
        println(s"connect $pIndex db~~~~")

        override def hasNext: Boolean = if (!pIter.hasNext) {
          println("close db~~~~");
          false
        } else true

        override def next(): String = {
          val str: String = pIter.next()
          println(s"$str select....")
          str + " selected"
        }
      }
    }).foreach(println)


  }

}
